//! 集群节点管理模块
//! 
//! 定义了集群中节点的抽象和管理功能

use crate::cluster::traits::{
    NodeInfo, NodeManager as NodeManagerTrait, NodeRole, NodeState
};
use crate::cluster::{ClusterError, ClusterResult};
use async_trait::async_trait;
use bit_vec::BitVec;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::time::{Duration, Instant};

/// 节点标志位
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Default)]
pub struct NodeFlags {
    /// 是否为主节点
    pub master: bool,
    /// 是否为从节点
    pub slave: bool,
    /// 是否处于故障状态
    pub fail: bool,
    /// 是否可能故障
    pub pfail: bool,
    /// 是否在握手中
    pub handshake: bool,
    /// 是否没有地址
    pub noaddr: bool,
}


/// 集群节点信息
#[derive(Debug, Clone)]
pub struct ClusterNode {
    /// 节点唯一标识符 (40字符十六进制字符串)
    pub id: String,
    
    /// 节点网络地址
    pub addr: SocketAddr,
    
    /// 节点角色
    pub role: NodeRole,
    
    /// 节点状态
    pub state: NodeState,
    
    /// 节点标志位
    pub flags: NodeFlags,
    
    /// 管理的槽位 (使用位向量表示)
    pub slots: BitVec,
    
    /// 最后ping时间
    pub last_ping: Instant,
    
    /// 最后pong时间
    pub last_pong: Instant,
    
    /// 故障计数
    pub fail_count: u32,
    
    /// 配置纪元 (用于故障恢复和选举)
    pub config_epoch: u64,
    
    /// 主节点ID (如果是从节点)
    pub master_id: Option<String>,
    
    /// 从节点列表 (如果是主节点)
    pub slaves: Vec<String>,
    
    /// 节点创建时间
    pub created_at: Instant,
    
    /// 节点元数据
    pub metadata: HashMap<String, String>,
}

impl ClusterNode {
    /// 创建新的集群节点
    pub fn new(id: String, addr: SocketAddr, role: NodeRole) -> Self {
        let now = Instant::now();
        let mut flags = NodeFlags::default();
        
        match role {
            NodeRole::Master => flags.master = true,
            NodeRole::Slave(_) => flags.slave = true,
            NodeRole::Sentinel => {}
        }
        
        Self {
            id,
            addr,
            role,
            state: NodeState::Online,
            flags,
            slots: BitVec::from_elem(16384, false), // 16384个槽位
            last_ping: now,
            last_pong: now,
            fail_count: 0,
            config_epoch: 0,
            master_id: None,
            slaves: Vec::new(),
            created_at: now,
            metadata: HashMap::new(),
        }
    }
    
    /// 创建主节点
    pub fn new_master(id: String, addr: SocketAddr) -> Self {
        Self::new(id, addr, NodeRole::Master)
    }
    
    /// 创建从节点
    pub fn new_slave(id: String, addr: SocketAddr, master_id: String) -> Self {
        let mut node = Self::new(id, addr, NodeRole::Slave(master_id.clone()));
        node.master_id = Some(master_id);
        node
    }
    
    /// 检查节点是否为主节点
    pub fn is_master(&self) -> bool {
        matches!(self.role, NodeRole::Master)
    }
    
    /// 检查节点是否为从节点
    pub fn is_slave(&self) -> bool {
        matches!(self.role, NodeRole::Slave(_))
    }
    
    /// 检查节点是否在线
    pub fn is_online(&self) -> bool {
        matches!(self.state, NodeState::Online)
    }
    
    /// 检查节点是否故障
    pub fn is_failed(&self) -> bool {
        matches!(self.state, NodeState::Fail) || self.flags.fail
    }
    
    /// 设置节点状态
    pub fn set_state(&mut self, state: NodeState) {
        let state_clone = state.clone();
        self.state = state;
        match state_clone {
            NodeState::Fail => self.flags.fail = true,
            NodeState::Suspect => self.flags.pfail = true,
            NodeState::Online => {
                self.flags.fail = false;
                self.flags.pfail = false;
                self.fail_count = 0;
            }
            _ => {}
        }
    }
    
    /// 添加槽位
    pub fn add_slot(&mut self, slot: u16) {
        if slot < 16384 {
            self.slots.set(slot as usize, true);
        }
    }
    
    /// 移除槽位
    pub fn remove_slot(&mut self, slot: u16) {
        if slot < 16384 {
            self.slots.set(slot as usize, false);
        }
    }
    
    /// 检查是否拥有槽位
    pub fn has_slot(&self, slot: u16) -> bool {
        if slot < 16384 {
            self.slots.get(slot as usize).unwrap_or(false)
        } else {
            false
        }
    }
    
    /// 获取拥有的槽位数量
    pub fn slot_count(&self) -> usize {
        self.slots.iter().filter(|&bit| bit).count()
    }
    
    /// 获取所有拥有的槽位
    pub fn get_slots(&self) -> Vec<u16> {
        self.slots
            .iter()
            .enumerate()
            .filter_map(|(i, bit)| if bit { Some(i as u16) } else { None })
            .collect()
    }
    
    /// 批量设置槽位
    pub fn set_slots(&mut self, slots: &[u16]) {
        // 先清空所有槽位
        self.slots.clear();
        self.slots.grow(16384, false);
        
        // 设置指定槽位
        for &slot in slots {
            self.add_slot(slot);
        }
    }
    
    /// 更新ping时间
    pub fn update_ping(&mut self) {
        self.last_ping = Instant::now();
    }
    
    /// 更新pong时间
    pub fn update_pong(&mut self) {
        self.last_pong = Instant::now();
        // 收到pong说明节点正常，重置故障计数
        if self.fail_count > 0 {
            self.fail_count = 0;
        }
    }
    
    /// 增加故障计数
    pub fn increment_fail_count(&mut self) {
        self.fail_count += 1;
    }
    
    /// 检查是否超时
    pub fn is_timeout(&self, timeout: Duration) -> bool {
        self.last_pong.elapsed() > timeout
    }
    
    /// 添加从节点
    pub fn add_slave(&mut self, slave_id: String) {
        if self.is_master() && !self.slaves.contains(&slave_id) {
            self.slaves.push(slave_id);
        }
    }
    
    /// 移除从节点
    pub fn remove_slave(&mut self, slave_id: &str) {
        if self.is_master() {
            self.slaves.retain(|id| id != slave_id);
        }
    }
    
    /// 将节点提升为主节点
    pub fn promote_to_master(&mut self) {
        self.role = NodeRole::Master;
        self.flags.master = true;
        self.flags.slave = false;
        self.master_id = None;
        self.config_epoch += 1;
    }
    
    /// 将节点降级为从节点
    pub fn demote_to_slave(&mut self, master_id: String) {
        self.role = NodeRole::Slave(master_id.clone());
        self.flags.master = false;
        self.flags.slave = true;
        self.master_id = Some(master_id);
        // 清空槽位和从节点列表
        self.slots.clear();
        self.slots.grow(16384, false);
        self.slaves.clear();
    }
    
    /// 设置元数据
    pub fn set_metadata<K: Into<String>, V: Into<String>>(&mut self, key: K, value: V) {
        self.metadata.insert(key.into(), value.into());
    }
    
    /// 获取元数据
    pub fn get_metadata(&self, key: &str) -> Option<&String> {
        self.metadata.get(key)
    }
    
    /// 生成节点信息字符串 (类似Redis CLUSTER NODES格式)
    pub fn to_info_string(&self) -> String {
        let role_str = match &self.role {
            NodeRole::Master => "master".to_string(),
            NodeRole::Slave(master_id) => format!("slave {master_id}"),
            NodeRole::Sentinel => "sentinel".to_string(),
        };
        
        let state_str = match self.state {
            NodeState::Online => "connected",
            NodeState::Offline => "disconnected",
            NodeState::Suspect => "suspect",
            NodeState::Fail => "fail",
            NodeState::Handshake => "handshake",
        };
        
        let slots_str = if self.is_master() && self.slot_count() > 0 {
            let slots = self.get_slots();
            let mut ranges = Vec::new();
            let mut start = slots[0];
            let mut end = slots[0];
            
            for &slot in slots.iter().skip(1) {
                if slot == end + 1 {
                    end = slot;
                } else {
                    if start == end {
                        ranges.push(start.to_string());
                    } else {
                        ranges.push(format!("{start}-{end}"));
                    }
                    start = slot;
                    end = slot;
                }
            }
            
            if start == end {
                ranges.push(start.to_string());
            } else {
                ranges.push(format!("{start}-{end}"));
            }
            
            ranges.join(" ")
        } else {
            String::new()
        };
        
        format!(
            "{} {}:{} {} {} {} {}",
            self.id,
            self.addr.ip(),
            self.addr.port(),
            role_str,
            state_str,
            self.config_epoch,
            slots_str
        ).trim().to_string()
    }
}

/// 节点管理器
#[derive(Debug)]
pub struct ClusterNodeManager {
    /// 本地节点
    local_node: ClusterNode,
    
    /// 已知节点映射
    nodes: HashMap<String, ClusterNode>,
    
    /// 故障超时时间
    fail_timeout: Duration,
}

impl ClusterNodeManager {
    /// 创建新的节点管理器
    pub fn new(local_node: ClusterNode, fail_timeout: Duration) -> Self {
        let mut nodes = HashMap::new();
        nodes.insert(local_node.id.clone(), local_node.clone());
        
        Self {
            local_node,
            nodes,
            fail_timeout,
        }
    }
    
    /// 获取本地节点
    pub fn local_node(&self) -> &ClusterNode {
        &self.local_node
    }
    
    /// 获取本地节点(可变)
    pub fn local_node_mut(&mut self) -> &mut ClusterNode {
        &mut self.local_node
    }
    
    /// 添加节点
    pub fn add_node(&mut self, node: ClusterNode) {
        self.nodes.insert(node.id.clone(), node);
    }
    
    /// 移除节点
    pub fn remove_node(&mut self, node_id: &str) -> Option<ClusterNode> {
        self.nodes.remove(node_id)
    }
    
    /// 获取节点
    pub fn get_node(&self, node_id: &str) -> Option<&ClusterNode> {
        self.nodes.get(node_id)
    }
    
    /// 获取节点(可变)
    pub fn get_node_mut(&mut self, node_id: &str) -> Option<&mut ClusterNode> {
        self.nodes.get_mut(node_id)
    }
    
    /// 获取所有节点
    pub fn all_nodes(&self) -> &HashMap<String, ClusterNode> {
        &self.nodes
    }
    
    /// 获取主节点列表
    pub fn master_nodes(&self) -> Vec<&ClusterNode> {
        self.nodes.values().filter(|node| node.is_master()).collect()
    }
    
    /// 获取从节点列表
    pub fn slave_nodes(&self) -> Vec<&ClusterNode> {
        self.nodes.values().filter(|node| node.is_slave()).collect()
    }
    
    /// 获取在线节点列表
    pub fn online_nodes(&self) -> Vec<&ClusterNode> {
        self.nodes.values().filter(|node| node.is_online()).collect()
    }
    
    /// 检查节点超时并更新状态
    pub fn check_timeouts(&mut self) -> Vec<String> {
        let mut failed_nodes = Vec::new();
        
        for (node_id, node) in self.nodes.iter_mut() {
            if node.is_timeout(self.fail_timeout) && !node.is_failed() {
                node.set_state(NodeState::Fail);
                failed_nodes.push(node_id.clone());
            }
        }
        
        failed_nodes
    }
    
    /// 获取节点数量统计
    pub fn node_stats(&self) -> (usize, usize, usize, usize) {
        let total = self.nodes.len();
        let masters = self.master_nodes().len();
        let slaves = self.slave_nodes().len();
        let online = self.online_nodes().len();
        
        (total, masters, slaves, online)
    }
}

// =============================================================================
// Trait实现
// =============================================================================

impl NodeInfo for ClusterNode {
    fn id(&self) -> &str {
        &self.id
    }
    
    fn addr(&self) -> SocketAddr {
        self.addr
    }
    
    fn role(&self) -> &NodeRole {
        &self.role
    }
    
    fn state(&self) -> &NodeState {
        &self.state
    }
    
    fn is_master(&self) -> bool {
        self.is_master()
    }
    
    fn is_slave(&self) -> bool {
        self.is_slave()
    }
    
    fn is_online(&self) -> bool {
        self.is_online()
    }
    
    fn is_failed(&self) -> bool {
        self.is_failed()
    }
    
    fn set_state(&mut self, state: NodeState) {
        self.set_state(state);
    }
    
    fn get_slots(&self) -> Vec<u16> {
        self.get_slots()
    }
    
    fn add_slot(&mut self, slot: u16) {
        self.add_slot(slot);
    }
    
    fn remove_slot(&mut self, slot: u16) {
        self.remove_slot(slot);
    }
    
    fn has_slot(&self, slot: u16) -> bool {
        self.has_slot(slot)
    }
}



#[async_trait]
impl NodeManagerTrait for ClusterNodeManager {
    type Node = ClusterNode;
    
    async fn create_node(&mut self, id: String, addr: SocketAddr, role: NodeRole) -> ClusterResult<Self::Node> {
        Ok(ClusterNode::new(id, addr, role))
    }
    
    async fn get_node(&self, id: &str) -> Option<&Self::Node> {
        self.nodes.get(id)
    }
    
    async fn get_node_mut(&mut self, id: &str) -> Option<&mut Self::Node> {
        self.nodes.get_mut(id)
    }
    
    async fn add_node(&mut self, node: Self::Node) -> ClusterResult<()> {
        let id = node.id.clone();
        if self.nodes.contains_key(&id) {
            return Err(ClusterError::Config(format!("节点 {id} 已存在")));
        }
        self.nodes.insert(id, node);
        Ok(())
    }
    
    async fn remove_node(&mut self, id: &str) -> ClusterResult<Option<Self::Node>> {
        Ok(self.nodes.remove(id))
    }
    
    async fn get_all_nodes(&self) -> Vec<&Self::Node> {
        self.nodes.values().collect()
    }
    
    async fn online_node_count(&self) -> usize {
        self.nodes.values().filter(|node| node.is_online()).count()
    }
    
    async fn contains_node(&self, id: &str) -> bool {
        self.nodes.contains_key(id)
    }
    
    async fn update_node_state(&mut self, id: &str, state: NodeState) -> ClusterResult<()> {
        match self.nodes.get_mut(id) {
            Some(node) => {
                node.set_state(state);
                Ok(())
            }
            None => Err(ClusterError::NodeNotFound(id.to_string()))
        }
    }
}

